package com.example.capsule_kafka.service;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.example.capsule_kafka.bean.GroupMessageBean;
import com.example.capsule_kafka.configuration.KafkaConstant;
import com.example.capsule_kafka.global.GlobalThreadPool;
import com.example.capsule_kafka.mapper.GroupMessageMapper;
import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author: zhuzhiming
 * @Date: 2022/5/4 0:04
 */
@Service
@Slf4j
public class GroupMessageService {

    @Resource
    GroupMessageMapper groupMessageMapper;

    @Autowired
    KafkaTemplate<String, String> kafkaTemplate;

    public String getMessages(String roomName) {

        List<GroupMessageBean> groupMessageBeanList = groupMessageMapper.getMessagesList(roomName);

        if (groupMessageBeanList == null || groupMessageBeanList.size() == 0) {
            return "";
        }

        if (groupMessageBeanList.size() > 300) {
            ThreadPoolExecutor poolExecutor = GlobalThreadPool.buildThreadPool();
            List<Integer> messageIds = new ArrayList<>();
            poolExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    int size = groupMessageBeanList.size();
                    int count = 0;
                    while (size > 300) {
                        messageIds.add(groupMessageBeanList.get(count).getId());
                        size--;
                        count++;
                    }
                    kafkaTemplate.send(KafkaConstant.DELETE_MESSAGE_TOPIC, new Gson().toJson(messageIds)).addCallback(new ListenableFutureCallback<SendResult<String, String>>() {

                        @Override
                        public void onFailure(Throwable ex) {
                            log.error("send delete message failure");
                        }

                        @Override
                        public void onSuccess(SendResult<String, String> result) {
                            log.info("send delete message success");
                        }
                    });
                }
            });
        }

        return new Gson().toJson(groupMessageBeanList);

    }

    public void updateMessage(Integer senderId, String avatarUrl, String avatarServerPath,String senderName) {

        QueryWrapper<GroupMessageBean> queryWrapper = new QueryWrapper<GroupMessageBean>().eq("sender_id", String.valueOf(senderId));

        List<GroupMessageBean> messageBeanList = groupMessageMapper.selectList(queryWrapper);

        for (GroupMessageBean groupMessageBean : messageBeanList) {
            groupMessageBean.setSenderAvatarLocalPath(avatarUrl);
            groupMessageBean.setSenderAvatarServerPath(avatarServerPath);
            groupMessageBean.setSenderName(senderName);
            groupMessageMapper.updateById(groupMessageBean);
        }

    }
}
